package com.ddxz.flink

import org.apache.flink.api.common.typeinfo.BasicTypeInfo
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat
import org.apache.flink.api.java.typeutils.RowTypeInfo
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * 数据库中数据如果采用流式处理方式，会导致结果不稳定。有时显示所有内容，有时打印部分内容，有时不打印任何内容
 */

object Word {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val types = new RowTypeInfo(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO,
      BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO)
    val t2 = new RowTypeInfo(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO)
    val ds1 = JDBCInputFormat.buildJDBCInputFormat()
      .setDBUrl("jdbc:mysql://192.168.12.140:3306/demo?useUnicode=true&characterEncoding=utf8&autoReconnect=true&failOverReadOnly=false&serverTimezone=Asia/Shanghai")
      .setDrivername("com.mysql.cj.jdbc.Driver")
      .setUsername("root")
      .setPassword("root")
      .setQuery("SELECT c_id, c_user, c_status, c_create_time, c_real_name, c_schema from t_meta_object")
      .setRowTypeInfo(types)
      .finish()
    ds1.openInputFormat()

    val ds2 = JDBCInputFormat.buildJDBCInputFormat()
      .setDBUrl("jdbc:mysql://192.168.12.140:3306/demo?useUnicode=true&characterEncoding=utf8&autoReconnect=true&failOverReadOnly=false&serverTimezone=Asia/Shanghai")
      .setDrivername("com.mysql.cj.jdbc.Driver")
      .setUsername("root")
      .setPassword("root")
      .setQuery("select c_id, c_rid, c_real_name, c_display_name from t_meta_column")
      .setRowTypeInfo(t2)
      .finish()
    ds2.openInputFormat()
    val s1 = env.createInput(ds1)
    val s2 = env.createInput(ds2)
//    s1.print()
//    s2.print()

    s1.join(s2)
      .where(t1 => t1.getField(0))
      .equalTo(t2 => t2.getField(1))
//      .window(new CountWindow(5))
      .window(TumblingProcessingTimeWindows.of(Time.milliseconds(1)))
//      .apply(DdxzJoin)
      .apply((t1, t2) => (t1.getField(3),
        t1.getField(4), t1.getField(5),
        t2.getField(2), t2.getField(3)))
      .print()
//    val v1 = env.createInput(ds1)(types)
//    val v2 = env.createInput(ds2)(t2)
//    v1.print()
//    v2.print()
//    var typeInfo = TypeInformation[Row]
//    v1.join(v2).where(i1=> i1.getField(0))(typeInfo)
//      .equalTo(i2=>i2.getField(1)).window(TumblingEventTimeWindows.of(Time.seconds(1)))
//      .apply((i1, i2) => (i1.getField(0), i1.getField(1), i1.getField(2), i1.getField(3),
//        i1.getField(4), i1.getField(5), i2.getField(0), i2.getField(1), i2.getField(2), i2.getField(3)))(typeInfo).print()
    ds2.closeInputFormat()
    ds1.closeInputFormat()
    env.execute("Word")
  }

}
